package com.project.base.common;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.SaveMode;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

public class SaveToHdfsUtils {

    private static final String BASE_DIR = Constants.getInstance().getProperty("hadoop.hdfs.base.dir");
    private static final String HDFS_URL = Constants.getInstance().getProperty("hadoop.hdfs.url");
    private static final SimpleDateFormat format = new SimpleDateFormat("yyyyMMdd");

    public static enum TABLE_NAME {
        API_REQUEST_TIME_DETAIL("请求时间"),
        LOG_DETAIL("日志详情"),
        REQUEST_SPECIAL_PARAM("请求特殊参数"),
        USER_ACCESS_PATH("请求的公共参数");
        private String cnName;

        private TABLE_NAME(String cnName) {
            this.setCnName(cnName);
        }

        public String getCnName() {
            return cnName;
        }

        public void setCnName(String cnName) {
            this.cnName = cnName;
        }
    }

    public static <T> void save(SQLContext sqlContext, TABLE_NAME tableName, List<T> dataList) {
        if (dataList == null || dataList.size() == 0) {
            return;
        }
        T data = dataList.get(0);
        DataFrame dataFrame = sqlContext.createDataFrame(dataList, data.getClass());
        save(dataFrame, tableName);
    }

    public static <T> void save(SQLContext sqlContext, TABLE_NAME tableName, JavaRDD<T> rdd) {
        T data = rdd.first();
        DataFrame dataFrame = sqlContext.createDataFrame(rdd, data.getClass());
        save(dataFrame, tableName);
    }

    private static void save(DataFrame dataFrame, TABLE_NAME tableName) {
        String fileName = getFileUri(tableName);
        dataFrame.write().mode(SaveMode.Append).save(fileName);
        System.out.println("数据保存完成：" + fileName);
    }

    private static String getFileUri(TABLE_NAME tableName) {
        return getFileUri(tableName, new Date());
    }

    public static String getFileUri(TABLE_NAME tableName, Date date) {
        StringBuffer sb = new StringBuffer();
        sb.append(HDFS_URL).append(BASE_DIR);
        sb.append("/").append(format.format(date));
        sb.append("/").append(tableName.name());
        sb.append("/data.parquet");
        return sb.toString();
    }
}
